package stream;

import com.ibm.tspaces.Field;
import com.ibm.tspaces.Tuple;
import com.ibm.tspaces.TupleSpace;
import com.ibm.tspaces.TupleSpaceException;

public class Stream<T extends java.io.Serializable> {
  static final String HEAD = "head";
  static final String TAIL = "tail";
  
  final Class<T>   elementType;
  final Tuple      headTemplate;
  final Tuple      tailTemplate;
  
  String     name;
  TupleSpace space;
  Tuple      tuple;
  
  public Stream(String name, Class<T> type, TupleSpace space) 
  throws TupleSpaceException {
    this.name  = name;
    this.space = space;
    elementType = type;
    headTemplate = new Tuple(name, HEAD, new Field(elementType));
    tailTemplate = new Tuple(name, TAIL, new Field(elementType));
  }
  
  public void init() 
  throws TupleSpaceException {
    space.write(new Tuple(name, HEAD, new Integer(0)));
    space.write(new Tuple(name, TAIL, new Integer(0)));
  }
  
  public void append(T val) 
  throws TupleSpaceException {
    tuple = space.waitToTake(tailTemplate);
    
    int index = (Integer) tuple.getField(2).getValue();
    tuple = new Tuple(name, TAIL, index + 1);
    space.write(tuple);
    
    tuple = new Tuple(name, index, val);
    space.write(tuple);
  }
  
  public T remove() 
  throws TupleSpaceException {
    tuple = space.waitToTake(headTemplate);
    
    int index = (Integer) tuple.getField(2).getValue();
    tuple = new Tuple(name, HEAD, index + 1);
    space.write(tuple);
    
    tuple = new Tuple(name, index, new Field(elementType));
    tuple = space.waitToTake(tuple);
    
    return (T) tuple.getField(2).getValue();
  }
}
